package com.chenzhiling.study.datasource

import com.chenzhiling.study.datasource.RemoteFile.RemoteFilePartition
import com.chenzhiling.study.util.FileUtil
import com.chenzhiling.study.util.PropertiesUtil.getString
import org.apache.spark.Partition
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{BinaryType, LongType, StringType, StructField, StructType}

import scala.collection.mutable.ArrayBuffer

/**
 * @Author: CHEN ZHI LING
 * @Date: 2021/8/11
 * @Description: 自定义文件schema
 */
object FileSchema {

  private val SPLIT_NUMBER: Int = getString("file.split.number").toInt
  private val FILE_NAME = "fileName"
  private val PATH = "path"
  private val SUFFIX = "suffix"
  private val SIZE = "size"
  private val COMMON_FILE = "commonFile"
  private val BYTE_ARRAY = "byteArray"

  /**
   * 自定义schema
   */
  val schema: StructType = StructType(
    StructField(FILE_NAME, StringType, nullable = true) ::
      StructField(PATH, StringType, nullable = true) ::
      StructField(SUFFIX, StringType, nullable = true) ::
      StructField(SIZE, LongType, nullable = true) ::
      StructField(COMMON_FILE,getSubSchema):: Nil)


  /**
   * 生成嵌套schema
   * @return
   */
  def getSubSchema: StructType={
    val fields = new ArrayBuffer[StructField]
    //根据切分份数，生成n个字段
    for (i <- 1 to SPLIT_NUMBER){
      fields.append(StructField(new StringBuilder(BYTE_ARRAY+i).toString(), BinaryType, nullable = false))
    }
    StructType(fields.toArray)
  }

  /**
   * 分区信息
   * @param path 路径
   * @return
   */
  def getPartition(path: String): Array[Partition]={
    val arrays = new Array[Partition](1)
    arrays(0) =  RemoteFilePartition(0, path = path)
    arrays
  }

  /**
   * 填充schema
   * @param fileName 文件名
   * @param path 路径
   * @param suffix 后缀
   * @param size 大小
   * @param array 内容字节数组
   * @return
   */
  def filledSchema(fileName:String,path:String,suffix:String,size:Long,array:Array[Byte]): Row ={
    val list: List[Array[Byte]] = FileUtil.splitArrayByNumber(array, SPLIT_NUMBER)
    val byteContent: Row = Row.fromSeq(list)
    Row(fileName, path, suffix, size, byteContent)
  }
}
